/*
 * Copyright 1999-2018 Alibaba Group Holding Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.alibaba.nacos.common.notify;

import com.alibaba.nacos.api.exception.runtime.NacosRuntimeException;
import com.alibaba.nacos.common.JustForTest;
import com.alibaba.nacos.common.notify.listener.SmartSubscriber;
import com.alibaba.nacos.common.notify.listener.Subscriber;
import com.alibaba.nacos.common.spi.NacosServiceLoader;
import com.alibaba.nacos.common.utils.ClassUtils;
import com.alibaba.nacos.common.utils.MapUtil;
import com.alibaba.nacos.common.utils.ThreadUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collection;
import java.util.Iterator;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

import static com.alibaba.nacos.api.exception.NacosException.SERVER_ERROR;

/**
 * 通知中心.
 * <p>
 * 主要用于注册发布者、调用发布者发布事件、为发布者注册订阅者、为指定的事件增加指定的订阅者等操作。
 * 可以说它完全接管了订阅者、发布者和事件他们的组合过程。
 * 直接调用通知中心的相关方法即可实现事件发布订阅者注册等功能
 * </p>
 *
 * @author <a href="mailto:liaochuntao@live.com">liaochuntao</a>
 * @author zongtanghu
 */
public class NotifyCenter {

    private static final Logger LOGGER = LoggerFactory.getLogger(NotifyCenter.class);

    /**
     * 单事件发布者内部的事件队列初始容量.
     */
    public static int ringBufferSize;

    /**
     * 多事件发布者内部的事件队列初始容量.
     */
    public static int shareBufferSize;

    /**
     * 发布者的状态.
     */
    private static final AtomicBoolean CLOSED = new AtomicBoolean(false);

    /**
     * 构造发布者的工厂.
     */
    private static final EventPublisherFactory DEFAULT_PUBLISHER_FACTORY;

    /**
     * 通知中心的实例.
     */
    private static final NotifyCenter INSTANCE = new NotifyCenter();

    /**
     * 默认的多事件发布者.
     */
    private DefaultSharePublisher sharePublisher;

    /**
     * 默认的单事件发布者类型.
     * 此处并未直接指定单事件发布者是谁，只是限定了它的类别
     * 因为单事件发布者一个发布者只负责一个事件，因此会存在
     * 多个发布者实例，后面按需创建，并缓存在publisherMap
     */
    private static Class<? extends EventPublisher> clazz;

    /**
     * 单事件发布者存储容器.
     */
    private final Map<String, EventPublisher> publisherMap = new ConcurrentHashMap<>(16);

    /*
     * 在静态代码块中主要就做了两件事:
     * 1.初始化单事件发布者：可以由用户扩展指定（通过Nacos SPI机制），也可以是Nacos默认的（DefaultPublisher）。
     * 2.初始化多事件发布者：DefaultSharePublisher。
     */
    static {
        // 初始化DefaultPublisher的queue容量值
        String ringBufferSizeProperty = "nacos.core.notify.ring-buffer-size";
        ringBufferSize = Integer.getInteger(ringBufferSizeProperty, 16384);

        // 初始化DefaultSharePublisher的queue容量值
        String shareBufferSizeProperty = "nacos.core.notify.share-buffer-size";
        shareBufferSize = Integer.getInteger(shareBufferSizeProperty, 1024);

        // 使用Nacos SPI机制获取事件发布者
        final Collection<EventPublisher> publishers = NacosServiceLoader.load(EventPublisher.class);
        // 获取迭代器
        Iterator<EventPublisher> iterator = publishers.iterator();
        if (iterator.hasNext()) {
            clazz = iterator.next().getClass();
        } else {
            // 若为空，则使用默认的发布器（单事件发布者）
            clazz = DefaultPublisher.class;
        }

        // 声明发布者工厂为一个函数，用于创建发布者实例
        DEFAULT_PUBLISHER_FACTORY = (cls, buffer) -> {
            try {
                EventPublisher publisher = clazz.newInstance();
                publisher.init(cls, buffer);
                return publisher;
            } catch (Throwable ex) {
                LOGGER.error("Service class newInstance has error : ", ex);
                throw new NacosRuntimeException(SERVER_ERROR, ex);
            }
        };

        try {
            // 初始化多事件发布者
            INSTANCE.sharePublisher = new DefaultSharePublisher();
            INSTANCE.sharePublisher.init(SlowEvent.class, shareBufferSize);
        } catch (Throwable ex) {
            LOGGER.error("Service class newInstance has error : ", ex);
        }
        // 增加关闭钩子，用于关闭Publisher
        ThreadUtils.addShutdownHook(NotifyCenter::shutdown);
    }

    @JustForTest
    public static Map<String, EventPublisher> getPublisherMap() {
        return INSTANCE.publisherMap;
    }

    @JustForTest
    public static EventPublisher getPublisher(Class<? extends Event> topic) {
        if (ClassUtils.isAssignableFrom(SlowEvent.class, topic)) {
            return INSTANCE.sharePublisher;
        }
        return INSTANCE.publisherMap.get(topic.getCanonicalName());
    }

    @JustForTest
    public static EventPublisher getSharePublisher() {
        return INSTANCE.sharePublisher;
    }

    /**
     * 关机.
     */
    public static void shutdown() {
        if (!CLOSED.compareAndSet(false, true)) {
            return;
        }
        LOGGER.warn("[NotifyCenter] Start destroying Publisher");

        for (Map.Entry<String, EventPublisher> entry : INSTANCE.publisherMap.entrySet()) {
            try {
                EventPublisher eventPublisher = entry.getValue();
                eventPublisher.shutdown();
            } catch (Throwable e) {
                LOGGER.error("[EventPublisher] shutdown has error : ", e);
            }
        }

        try {
            INSTANCE.sharePublisher.shutdown();
        } catch (Throwable e) {
            LOGGER.error("[SharePublisher] shutdown has error : ", e);
        }

        LOGGER.warn("[NotifyCenter] Destruction of the end");
    }

    /**
     * 注册订阅者.
     *
     * @param consumer subscriber
     */
    public static void registerSubscriber(final Subscriber consumer) {
        registerSubscriber(consumer, DEFAULT_PUBLISHER_FACTORY);
    }

    /**
     * 注册订阅者.
     *
     * @param consumer subscriber
     * @param factory  publisher factory.
     */
    public static void registerSubscriber(final Subscriber consumer, final EventPublisherFactory factory) {
        // 若想监听多个事件，实现SmartSubscriber.subscribeTypes()方法，在里面返回多个事件的列表即可
        // 多事件订阅者注册
        if (consumer instanceof SmartSubscriber) {
            // 获取事件列表
            for (Class<? extends Event> subscribeType : ((SmartSubscriber) consumer).subscribeTypes()) {
                // 判断它的事件类型来决定采用哪种Publisher，多事件订阅者由多事件发布者调度
                if (ClassUtils.isAssignableFrom(SlowEvent.class, subscribeType)) {
                    // 注册到多事件发布者中
                    INSTANCE.sharePublisher.addSubscriber(consumer, subscribeType);
                } else {
                    // 注册到单事件发布者中
                    addSubscriber(consumer, subscribeType, factory);
                }
            }
            return;
        }

        // 单事件的订阅者注册
        final Class<? extends Event> subscribeType = consumer.subscribeType();
        // 防止误使用，万一有人在使用单事件订阅者Subscriber的时候传入了SlowEvent则可以在此避免
        if (ClassUtils.isAssignableFrom(SlowEvent.class, subscribeType)) {
            INSTANCE.sharePublisher.addSubscriber(consumer, subscribeType);
            return;
        }

        // 注册到单事件发布者中
        addSubscriber(consumer, subscribeType, factory);
    }

    /**
     * 单事件发布者添加订阅者.
     *
     * @param consumer      subscriber instance.
     * @param subscribeType subscribeType.
     * @param factory       publisher factory.
     */
    private static void addSubscriber(final Subscriber consumer, Class<? extends Event> subscribeType, EventPublisherFactory factory) {
        // 获取类的规范名称，实际上就是包名加类名，作为topic
        final String topic = ClassUtils.getCanonicalName(subscribeType);
        synchronized (NotifyCenter.class) {
            /*
             * 生成指定类型的发布者，并将其放入publisherMap中
             * 使用topic为key从publisherMap获取数据，若为空则使用publisherFactory函数并传递subscribeType和ringBufferSize来实例
             * 化一个clazz类型的发布者对象，使用topic为key放入publisherMap中，实际上就是为每一个类型的事件创建一个发布者。具体
             * 可查看publisherFactory的逻辑。
             */
            MapUtil.computeIfAbsent(INSTANCE.publisherMap, topic, factory, subscribeType, ringBufferSize);
        }
        // 获取生成的发布者对象，将订阅者添加进去
        EventPublisher publisher = INSTANCE.publisherMap.get(topic);
        if (publisher instanceof ShardedEventPublisher) {
            ((ShardedEventPublisher) publisher).addSubscriber(consumer, subscribeType);
        } else {
            publisher.addSubscriber(consumer);
        }
    }

    /**
     * 注销事件发布者.
     *
     * @param consumer subscriber instance.
     */
    public static void deregisterSubscriber(final Subscriber consumer) {
        // 若是多事件订阅者
        if (consumer instanceof SmartSubscriber) {
            // 获取事件列表
            for (Class<? extends Event> subscribeType : ((SmartSubscriber) consumer).subscribeTypes()) {
                // 若是慢事件
                if (ClassUtils.isAssignableFrom(SlowEvent.class, subscribeType)) {
                    // 从多事件发布者中移除
                    INSTANCE.sharePublisher.removeSubscriber(consumer, subscribeType);
                } else {
                    // 从单事件发布者中移除
                    removeSubscriber(consumer, subscribeType);
                }
            }
            return;
        }

        // 若是单事件订阅者
        final Class<? extends Event> subscribeType = consumer.subscribeType();
        // 判断是否是慢事件
        if (ClassUtils.isAssignableFrom(SlowEvent.class, subscribeType)) {
            INSTANCE.sharePublisher.removeSubscriber(consumer, subscribeType);
            return;
        }

        // 调用移除方法
        if (removeSubscriber(consumer, subscribeType)) {
            return;
        }
        throw new NoSuchElementException("The subscriber has no event publisher");
    }

    /**
     * Remove subscriber.
     *
     * @param consumer      subscriber instance.
     * @param subscribeType subscribeType.
     * @return whether remove subscriber successfully or not.
     */
    private static boolean removeSubscriber(final Subscriber consumer, Class<? extends Event> subscribeType) {
        // 获取topic
        final String topic = ClassUtils.getCanonicalName(subscribeType);
        // 根据topic获取对应的发布者
        EventPublisher eventPublisher = INSTANCE.publisherMap.get(topic);
        if (null == eventPublisher) {
            return false;
        }
        // 从发布者中移除订阅者
        if (eventPublisher instanceof ShardedEventPublisher) {
            ((ShardedEventPublisher) eventPublisher).removeSubscriber(consumer, subscribeType);
        } else {
            eventPublisher.removeSubscriber(consumer);
        }
        return true;
    }

    /**
     * 发布事件.
     *
     * @param event class Instances of the event.
     */
    public static boolean publishEvent(final Event event) {
        try {
            return publishEvent(event.getClass(), event);
        } catch (Throwable ex) {
            LOGGER.error("There was an exception to the message publishing : ", ex);
            return false;
        }
    }

    /**
     * 发布事件.
     *
     * @param eventType class Instances type of the event type.
     * @param event     event instance.
     */
    private static boolean publishEvent(final Class<? extends Event> eventType, final Event event) {
        // 慢事件处理
        if (ClassUtils.isAssignableFrom(SlowEvent.class, eventType)) {
            return INSTANCE.sharePublisher.publish(event);
        }

        // 常规事件处理
        final String topic = ClassUtils.getCanonicalName(eventType);
        EventPublisher publisher = INSTANCE.publisherMap.get(topic);
        if (publisher != null) {
            return publisher.publish(event);
        }
        LOGGER.warn("There are no [{}] publishers for this event, please register", topic);
        return false;
    }

    /**
     * 注册发布者.
     *
     * @param eventType class Instances type of the event type.
     * @return share publisher instance.
     */
    public static EventPublisher registerToSharePublisher(final Class<? extends SlowEvent> eventType) {
        return INSTANCE.sharePublisher;
    }

    /**
     * 使用默认工厂注册发布者.
     *
     * @param eventType    事件类型.
     * @param queueMaxSize 队列最大值.
     */
    public static EventPublisher registerToPublisher(final Class<? extends Event> eventType, final int queueMaxSize) {
        return registerToPublisher(eventType, DEFAULT_PUBLISHER_FACTORY, queueMaxSize);
    }

    /**
     * 使用指定工厂注册发布者.
     *
     * @param eventType    事件类型.
     * @param factory      发布者工厂.
     * @param queueMaxSize 队列最大值.
     */
    public static EventPublisher registerToPublisher(final Class<? extends Event> eventType, final EventPublisherFactory factory, final int queueMaxSize) {
        // 慢事件由多事件发布者处理
        if (ClassUtils.isAssignableFrom(SlowEvent.class, eventType)) {
            return INSTANCE.sharePublisher;
        }
        // 若不是慢事件，因为它可以存在多个不同的类型，因此需要判断对应的发布者是否存在
        final String topic = ClassUtils.getCanonicalName(eventType);
        synchronized (NotifyCenter.class) {
            // 当前传入的事件类型对应的发布者，有则忽略无则新建
            MapUtil.computeIfAbsent(INSTANCE.publisherMap, topic, factory, eventType, queueMaxSize);
        }
        return INSTANCE.publisherMap.get(topic);
    }

    /**
     * Register publisher.
     *
     * @param eventType class Instances type of the event type.
     * @param publisher the specified event publisher
     */
    public static void registerToPublisher(final Class<? extends Event> eventType, final EventPublisher publisher) {
        if (null == publisher) {
            return;
        }
        final String topic = ClassUtils.getCanonicalName(eventType);
        synchronized (NotifyCenter.class) {
            INSTANCE.publisherMap.putIfAbsent(topic, publisher);
        }
    }

    /**
     * 注销发布者.
     *
     * @param eventType class Instances type of the event type.
     */
    public static void deregisterPublisher(final Class<? extends Event> eventType) {
        // 获取topic
        final String topic = ClassUtils.getCanonicalName(eventType);
        // 根据topic移除对应的发布者
        EventPublisher publisher = INSTANCE.publisherMap.remove(topic);
        try {
            // 调用关闭方法
            publisher.shutdown();
        } catch (Throwable ex) {
            LOGGER.error("There was an exception when publisher shutdown : ", ex);
        }
    }

}
